package com.shouke.RabbitMQ;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.rabbitmq.client.*;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;

public class RabbitMQClient {
    private static final Logger logger = LogManager.getLogger(RabbitMQClient.class);

    private String host;
    private Integer port;
    private String username;
    private String password;
    private Boolean isAutoRecoveryEnabled;
    private Integer networkRecoveryInterval;
    private String virtualHost;
    private Integer connectionTimeout;
    private Integer handshakeTimeout;
    private Integer shutdownTimeout;
    private Channel channel;
    private Connection conn;

    public  RabbitMQClient(String host,
                           Integer port,
                           String username,
                           String password,
                           Boolean isAutoRecoveryEnabled,
                           Integer networkRecoveryInterval,
                           String virtualHost,
                           Integer connectionTimeout,
                           Integer handshakeTimeout,
                           Integer shutdownTimeout){
        this.host = host;
        this.port = port;
        this.username = username;
        this.password = password;
        this.isAutoRecoveryEnabled = isAutoRecoveryEnabled;
        this.networkRecoveryInterval = networkRecoveryInterval;
        this.virtualHost = virtualHost;
        this.connectionTimeout = connectionTimeout;
        this.handshakeTimeout = handshakeTimeout;
        this.shutdownTimeout = shutdownTimeout;
        this.conn = null;
        this.channel = null;

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(this.host);
        factory.setPort(this.port);  // 默认端口为 5672
        factory.setUsername(this.username); // 设置连接登录用户
        factory.setPassword(this.password); // 设置用户访问密码
        factory.setAutomaticRecoveryEnabled(this.isAutoRecoveryEnabled);  // 开启Connection自动恢复功能，这意味着如果连接丢失，客户端将尝试重新连接到 RabbitMQ 服务器。
        factory.setNetworkRecoveryInterval(this.networkRecoveryInterval); // 尝试重连时间间隔 // 设置为 5000：如果RabbitMQ客户端失去连接后,每5秒自动尝试重连一次
        factory.setVirtualHost(this.virtualHost); // 设置虚拟主机，默认 /
        factory.setConnectionTimeout(this.connectionTimeout); // 设置TCP连接超时时间 默认 60000(60秒)
        factory.setHandshakeTimeout(this.handshakeTimeout); // 设置SSL握手超时时间 默认 10000(10秒)
        factory.setShutdownTimeout(this.shutdownTimeout); // 设置客户端关闭前等待操作完成的最大时间 默认 10000(10秒)

        try {
            this.conn = factory.newConnection();
            this.channel = this.conn.createChannel();
        } catch (Exception e){
            logger.error("Error occured while creating rabbitMQ channel: ", e);
        }

    }

    public void close() {
        try {
            if (this.channel != null) {
                this.channel.close();
            }
            if (this.conn != null) {
                this.conn.close();
            }

        }catch(Exception e) {
            logger.error("Error occured while closing rabbitMQ channel or connection: ", e);

        }
    }

    public String basicPublish(String exchange, String queueName, Boolean isCreateQueue, String routingKey, Map<String, Object>  messageHeaders, Map<String, String>  messageProperties, String dataType, String message) {
        String result = "";
        try {
            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();

            // MessageProperties.PERSISTENT_TEXT_PLAIN is a static instance of AMQP.BasicProperties
            // that contains a delivery mode and a priority. So we pass them to the builder.
            builder.deliveryMode(MessageProperties.PERSISTENT_TEXT_PLAIN.getDeliveryMode());
            builder.priority(MessageProperties.PERSISTENT_TEXT_PLAIN.getPriority());

            setBuilderProperties(builder, messageProperties);

            // Add the headers to the builder.
            builder.headers(messageHeaders);

            if (dataType.equals("JSON")){
                builder.contentType("application/json");
            }
            // Use the builder to create the BasicProperties object.
            AMQP.BasicProperties messageProps = builder.build();

            if (!queueName.isEmpty() && isCreateQueue) {
                this.channel.queueDeclare(queueName, false, false, false, null);
            }

            this.channel.basicPublish(exchange, routingKey, messageProps, message.getBytes(StandardCharsets.UTF_8));
            logger.debug(" [x] Sent '" + message + "'");
            result = "Publish Message SuccessFully";
        } catch (Exception e){
            logger.error("Error occured while executing rabbitMQ message basic publishing!", e);
            result = "Publish Message Fail: " + e.toString();
        }
        return result;
    }

    public void setBuilderProperties(AMQP.BasicProperties.Builder builder, Map<String, String>  messageProperties){
        for (String key : messageProperties.keySet()) {
            switch(key)
            {
                case "contentType" :
                    builder.contentType(messageProperties.get(key));
                    break;
                case "contentEncoding" :
                    builder.contentEncoding(messageProperties.get(key));
                    break;
                case "deliveryMode" :
                    builder.deliveryMode(Integer.parseInt(messageProperties.get(key)));
                    break;
                case "priority" :
                    builder.priority(Integer.parseInt(messageProperties.get(key)));
                    break;
                case "correlationId" :
                    builder.correlationId(messageProperties.get(key));
                    break;
                case "replyTo" :
                    builder.replyTo(messageProperties.get(key));
                    break;
                case "expiration" :
                    builder.expiration(messageProperties.get(key));
                    break;
                case "messageId" :
                    builder.messageId(messageProperties.get(key));
                    break;
                case "timestamp" :
                    builder.timestamp(new Date(Long.parseLong(messageProperties.get(key))));
                    break;
                case "type" :
                    builder.type(messageProperties.get(key));
                    break;
                case "userId" :
                    builder.userId(messageProperties.get(key));
                    break;
                case "appId" :
                    builder.appId(messageProperties.get(key));
                    break;
                case "clusterId" :
                    builder.clusterId(messageProperties.get(key));
            }
        }


    }

}